package com.xwj.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 反应器：多线程模式
 *
 * @author xwj
 * @date 2020/6/11
 */
public class MultiThreadEchoServerReactor {
    private ServerSocketChannel serverSocketChannel;
    private AtomicInteger next = new AtomicInteger(0);
    /**
     * 多个选择器
     */
    private Selector[] selectors = new Selector[2];

    private SubReactor[] subReactors = null;

    public MultiThreadEchoServerReactor() throws IOException {
        // 初始化多个选择器
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();

        serverSocketChannel = ServerSocketChannel.open();
//        serverSocketChannel.socket().bind(new InetSocketAddress(8888));
        serverSocketChannel.bind(new InetSocketAddress(8888));
        // 非阻塞
        serverSocketChannel.configureBlocking(false);
        // 第一个选择器负责监控连接事件
        SelectionKey sk = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);
        sk.attach(new AcceptorHandler());

        SubReactor subReactor1 = new SubReactor(selectors[0]);
        SubReactor subReactor2 = new SubReactor(selectors[1]);
        subReactors = new SubReactor[]{subReactor1, subReactor2};
    }

    public static void main(String[] args) throws IOException {
        new MultiThreadEchoServerReactor().startServer();
    }

    public void startServer() {
        new Thread(subReactors[0]).start();
        new Thread(subReactors[1]).start();
    }

    /**
     * 子反应器
     */
    class SubReactor implements Runnable {
        /**
         * 一个线程对应一个选择器
         */
        private Selector selector;

        public SubReactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select();
                    Iterator<SelectionKey> selectionKey = selector.selectedKeys().iterator();
                    while (selectionKey.hasNext()) {
                        // 反应器负责处理dispatch接收到的事件
                        SelectionKey sk = selectionKey.next();
                        dispatch(sk);
                        selectionKey.remove();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private void dispatch(SelectionKey sk) {
            Runnable runnable = (Runnable) sk.attachment();
            // 调用之前attach绑定到选择键的handler处理器对象
            if (runnable != null) {
                runnable.run();
            }
        }
    }

    /**
     * Handler：新连接处理器
     */
    class AcceptorHandler implements Runnable {

        @Override
        public void run() {
            try {
                SocketChannel sc = serverSocketChannel.accept();
                if (sc != null) {
                    new MultiThreadEchoHandler(selectors[next.get()], sc);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }
}
